package com.bff.gaia.optimizer.joinarrangement;

import com.bff.gaia.optimizer.traversals.IdAndEstimatesVisitor;

import java.util.ArrayList;
import java.util.Collections;
import java.util.HashSet;

public class DynamicPlan {
	private IdAndEstimatesVisitor.JoinTreeRecord joinTreeRecord;

	DynamicPlan(IdAndEstimatesVisitor.JoinTreeRecord joinTreeRecord) {
		this.joinTreeRecord = joinTreeRecord;
	}

	public class Tree implements Comparable<Tree> {
		Tree child1 = null;
		Tree child2 = null;
		private int id;
		private HashSet<Integer> nodeSet = new HashSet<>();
		private long estimatedNumRecords;
		private long estimatedOutputSize;
		private long cumulativeCost;

		Tree connectedTree;

		public int getId() {
			return id;
		}

		public long getEstimatedNumRecords() {
			return estimatedNumRecords;
		}

		public long getEstimatedOutputSize() {
			return estimatedOutputSize;
		}

		public Tree(HashSet<Integer> nodeSet, Tree child1, Tree child2, long estimatedNumRecords, long estimatedOutputSize) {
			this.nodeSet = nodeSet;
			this.child1 = child1;
			this.child2 = child2;
			this.estimatedNumRecords = estimatedNumRecords;
			this.estimatedOutputSize = estimatedOutputSize;
			cumulativeCost = transmissionCost(child1, child2);
		}

		public Tree(int id) {
			this.id = id;
			nodeSet.add(id);
			estimatedNumRecords = joinTreeRecord.getJoinTreeSourceNodeMap().get(id).getEstimatedNumRecords();
			estimatedOutputSize = joinTreeRecord.getJoinTreeSourceNodeMap().get(id).getEstimatedOutputSize();
			cumulativeCost = 0;
		}

		public int compareTo(Tree tree) {
			return (int) (this.cumulativeCost - tree.cumulativeCost);
		}

		@Override
		public String toString() {
			String str = "";
			if (child1 != null) {
				str += "child1:" + child1.nodeSet + " ";
			}
			if (child2 != null) {
				str += "child2:" + child2.nodeSet + " ";
			}
			str += cumulativeCost + "\t";
			str += joinTreeRecord.width(nodeSet);
			return str;
		}
	}

	Tree reArrange() {
		int nodeNum = joinTreeRecord.getJoinTreeSourceNodeMap().size();
		Tree[][] matrix = new Tree[nodeNum][nodeNum];
		for (int i = 0; i < nodeNum; i++) {
			for (int j = 0; j < nodeNum; j++) {
				if (i == 0) {
					matrix[i][j] = new Tree(j);
				} else {
					boolean flag = false;
					long cost = Long.MAX_VALUE, num = Long.MAX_VALUE;
					Tree tree1 = null, tree2 = null;
					for (int m = 0; m < i; m++) {
						for (int n = 0; n < nodeNum; n++) {
							if (isConnectable(matrix[m][n], matrix[i - m - 1][j]) && transmissionCost(matrix[m][n], matrix[i - m - 1][j]) < cost) {
								tree1 = matrix[m][n];
								tree2 = matrix[i - m - 1][j];
								num = estimateNum(tree1, tree2);
								cost = transmissionCost(tree1, tree2);
								flag = true;
							}
						}
					}
					if (flag && tree1.connectedTree != tree2) {
						HashSet<Integer> newSet = new HashSet<>();
						newSet.addAll(tree1.nodeSet);
						newSet.addAll(tree2.nodeSet);
						matrix[i][j] = new Tree(newSet, tree1, tree2, num, estimateSize(tree1, tree2));
						tree1.connectedTree = tree2;
						tree2.connectedTree = tree1;
					} else {
						matrix[i][j] = null;
					}
				}
			}
		}
		for (int i = 0; i < nodeNum; i++) {
			for (int j = 0; j < nodeNum; j++) {
				System.out.print((matrix[i][j] == null ? "null" : (matrix[i][j].cumulativeCost == 0 ? matrix[i][j].estimatedOutputSize : matrix[i][j].cumulativeCost)) + "\t");
			}
			System.out.print("\n");
		}

		ArrayList<Tree> result = new ArrayList<>();
		for (Tree tree : matrix[nodeNum - 1]) {
			if (tree != null) {
				result.add(tree);
			}
		}

		return Collections.min(result);
	}

	private long estimateNum(Tree tree1, Tree tree2) {
		return Math.max(tree1.estimatedNumRecords, tree2.estimatedNumRecords);
	}

	private long estimateSize(Tree tree1, Tree tree2) {
		return estimateNum(tree1, tree2) * width(tree1, tree2);
	}

	private int width(Tree tree1, Tree tree2) {
		HashSet<Integer> newSet = new HashSet<>();
		newSet.addAll(tree1.nodeSet);
		newSet.addAll(tree2.nodeSet);
		return joinTreeRecord.width(newSet);
	}

	private long transmissionCost(Tree tree1, Tree tree2) {
		return tree1.cumulativeCost + tree1.estimatedOutputSize + tree2.cumulativeCost + tree2.estimatedOutputSize;
	}

	private boolean isConnectable(Tree tree1, Tree tree2) {
		if (tree1 == null || tree2 == null) {
			return false;
		}
		for (int id : tree1.nodeSet) {
			if (tree2.nodeSet.contains(id)) {
				return false;
			}
		}
		return joinTreeRecord.isConnectable(tree1.nodeSet, tree2.nodeSet);
	}
}